package dev_ops.tools.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * 
 * @see <a href=
 *      "https://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html">090_KafkaProducer.html</a>
 *      <br/>
 *      https://github.com/mapr-demos/kafka-sample-programs/blob/master/src/main/java/com/mapr/examples/Consumer.java
 */
public class KafkaProducer_WR {
    public static void main(String[] args) {
        Properties props = Prop_env_Test();
        Producer<String, byte[]> producer = new KafkaProducer<String, byte[]>(props);
        for (int i = 0; i < 5; i++) {
            ProducerRecord<String, byte[]> record = newRecord_Test(
                    // "zhangxin-server", "user",
                    "ympns-push","test",
                    "{\"bizType\":\"user\",\"bizAction\":\"headImg\",\"m\":\"test msg\",\"request_time\":\"2017-10-19 18:40:39,258\",\"versionCode\":138}".getBytes());
            //record.
            try {
                System.out.println("send");
                producer.send(record);
                System.out.println("send ok");
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(producer.metrics().toString());
        }
        //producer.close();
    }

    public static Properties Prop_env_Test() {
        Properties props = newCommonEnv();
        props.put("bootstrap.servers", "10.136.24.76:9091");
        //props.put("bootstrap.servers", "10.136.24.76:9092");
        return props;
    }

    /**
     * common settings, better not change.
     * 
     * @return
     */
    public static Properties newCommonEnv() {
        Properties props = new Properties();
        props.put("acks", "1");
        props.put("retries", 2);
        props.put("request.timeout.ms", 10000);
        props.put("batch.size", 16384);
        props.put("linger.ms", 3);
        props.put("buffer.memory", 33554432);
        props.put("max.request.size", 2097152);
        props.put("batch.size", 2097152);
        props.put("metadata.fetch.timeout.ms", 5000); //15_000
        props.put("key.serializer", KafkaUtils.String_ser);
        props.put("value.serializer", KafkaUtils.ByteArray_ser);
        //props.put("partitioner.class", "dev_ops.tools.ForceRandomPartitioner");
        return props;
    }

    public static <R, S> ProducerRecord<R, S> newRecord_Test(String topic, R key, S msg) {
        ProducerRecord<R, S> record = new ProducerRecord<R, S>(topic, null, System.currentTimeMillis(), key, msg);
        return record;
    }

    /**
     * @see org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord,
     *      Callback)
     */
    // TODO
    public static void withCallBack() {
        // Fully non-blocking usage can make use of the Callback parameter to provide a
        // callback that will be invoked when the request is complete.
        //
        // ProducerRecord<byte[],byte[]> record = new
        // ProducerRecord<byte[],byte[]>("the-topic", key, value);
        // producer.send(myRecord,
        // new Callback() {
        // public void onCompletion(RecordMetadata metadata, Exception e) {
        // if(e != null)
        // e.printStackTrace();
        // System.out.println("The offset of the record we just sent is: " +
        // metadata.offset());
        // }
        // });
    }
}
